#pragma once

#include "stdafx.h"
#include "WorkerThreadPool.h"

using namespace ISAPI_Python::Win32Helpers;

#define DEFAULT_POOL_SIZE 255

void WorkerThreadPool::Init(DWORD poolSize) {
	if(poolSize <= 0) {
		PoolSize = DEFAULT_POOL_SIZE;
	} else {
		PoolSize = poolSize;
	}

	//The IO Completion Port is the backbone of an old-style
	// Win32 thread pool...it queues and dispatches work
	// to the pool. Even though it was originally designed for
	// completing asynchronous disk/network I/O operations.
	IoPort = CreateIoCompletionPort((HANDLE)INVALID_HANDLE_VALUE, NULL, 0, 0);

	PooledThreads = new HANDLE[PoolSize];

	//Create the threads for the thread pool manually.
	for(DWORD threadCounter = 0; threadCounter < PoolSize; threadCounter++) {
		PooledThreads[threadCounter] = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)&ThreadCallback, this, 0, NULL);
	}
}

WorkerThreadPool::WorkerThreadPool() {
	Init(0);
}

WorkerThreadPool::WorkerThreadPool(const DWORD poolSize) {
	Init(poolSize);
}

void WorkerThreadPool::QueueWorkItem(WORKER_THREAD_CALLBACK callback, PVOID callbackArgument) {
	//Use a structure to pass everything on the 64-bit compatible
	// argument.
	UserCallback* callbackStruct = new UserCallback();
	callbackStruct->callback = callback;
	callbackStruct->callbackArgument = callbackArgument;

	PostQueuedCompletionStatus(IoPort, 0, (ULONG_PTR)callbackStruct, NULL);
}

//static shim back into the OO world of C++, because Win32 isn't C++.
DWORD WorkerThreadPool::ThreadCallback(LPVOID lpThreadParameter) {
    WorkerThreadPool* currentThreadPool = (WorkerThreadPool*)lpThreadParameter;    
    return currentThreadPool->WorkerCallback();
}

DWORD WorkerThreadPool::WorkerCallback() {
	DWORD NumberOfBytesTransferred;
	ULONG_PTR CompletionKey;
	LPOVERLAPPED lpOverlapped; 

	//GetQueuedCompletionStatus() doesn't really care what you pass on its numeric
	// arguments, so even though they look like they have specific purposes,
	// they don't (in this particular instance).
	// They just get passed directly through from the queue.
	//
	//CompletionKey is begin used to transfer the callers callback information.
	//lpOverlapped == 0xFFFFFFFF is used for the "thread, please exit" signal.
	while(GetQueuedCompletionStatus(IoPort, &NumberOfBytesTransferred, &CompletionKey, &lpOverlapped, INFINITE)) {
		if((DWORD)lpOverlapped == 0xFFFFFFFF) {
			break;
		} else {
			UserCallback* callbackStruct = (UserCallback*)CompletionKey;

			if(callbackStruct != NULL) {
				callbackStruct->callback(callbackStruct->callbackArgument);
			}
		}

		if(CompletionKey != NULL) {
			UserCallback* callbackStruct = (UserCallback*)CompletionKey;
			delete callbackStruct;
		}
	}

	return 0;
}

WorkerThreadPool::~WorkerThreadPool() throw() {
	//Get rid of the pool's threads.
	for(DWORD threadCounter = 0; threadCounter < PoolSize; threadCounter++ ) {
		//I'm using lpOverlapped for a thread termination signal
		// so I don't have to be mean and call TerminateThread().
		PostQueuedCompletionStatus(IoPort, 0, NULL, (LPOVERLAPPED)0xFFFFFFFF);
	}	

	WaitForMultipleObjects(PoolSize, PooledThreads, TRUE, 120000);
	CloseHandle(IoPort);

	//Now that I *think* the threads have exited gracefully,
	// I'll get rid of the handles. It looks as though threads don't
	// release their memory until you close the handle as well.
	for(DWORD threadCounter = 0; threadCounter < PoolSize; threadCounter++) {
		CloseHandle(PooledThreads[threadCounter]);
	}

	delete PooledThreads;
	PooledThreads = NULL;
}